/* Copyright (c) 2021 OceanBase and/or its affiliates. All rights reserved.
miniob is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
         http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */

//
// Created by WangYunlai on 2021/6/9.
//

#include "sql/operator/update_physical_operator.h"
#include "common/log/log.h"
#include "common/type/data_type.h"
#include "sql/expr/expression.h"
#include "sql/stmt/insert_stmt.h"
#include "sql/stmt/update_stmt.h"
#include "storage/field/field_meta.h"
#include "storage/table/table.h"
#include "storage/trx/trx.h"
#include <cstdint>
#include <random>

using namespace std;

RC UpdatePhysicalOperator::open(Trx *trx)
{
  LOG_TRACE("---------------children_size:%d", children_.size());
  if (children_.empty()) {
    return RC::SUCCESS;
  }

  std::unique_ptr<PhysicalOperator> &child = children_[0];

  RC rc = child->open(trx);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to open child operator: %s", strrc(rc));
    return rc;
  }

  trx_ = trx;

  while (OB_SUCC(rc = child->next())) {

    Tuple *tuple = child->current_tuple();
    if (nullptr == tuple) {
      LOG_WARN("failed to get current record: %s", strrc(rc));
      return rc;
    }

    RowTuple *row_tuple = static_cast<RowTuple *>(tuple);
    // 可以利用row_tuple来取出一行中的所有值，对属性名字进行匹配，匹配成功的，修改为新的值
    std::vector<Value> cells;
    for (int i = 0; i < row_tuple->cell_num(); i++) {
      Value cell;
      row_tuple->cell_at(i, cell);
      cells.push_back(cell);
    }
    Record new_record;
    if (value_.size() == 0) {
      LOG_WARN("value not exist!");
      return RC::INVALID_ARGUMENT;
    }
    if (field_meta_.size() == 0) {
      LOG_WARN("field not exist!");
      return RC::SCHEMA_FIELD_NOT_EXIST;
    }
    LOG_TRACE("------------------Start to make new record!");
    vector<FieldMeta> field_meta;
    vector<Value>     value;
    for (unsigned i = 0; i < field_meta_.size(); i++) {
      FieldMeta reg_field_meta(field_meta_[i]->name(),
          field_meta_[i]->type(),
          field_meta_[i]->offset(),
          field_meta_[i]->len(),
          field_meta_[i]->visible(),
          field_meta_[i]->field_id(),
          field_meta_[i]->nullable());
      Value     reg_value;
      if (value_[i]->type() != ExprType::SUB_QUERY && value_[i]->type() != ExprType::VALUE) {
        LOG_WARN("INVALID_ARGUMENT!");
        return RC::INVALID_ARGUMENT;
      }
      rc = value_[i]->try_get_value(reg_value);
      if (rc != RC::SUCCESS && rc != RC::RECORD_EOF) {
        LOG_WARN("failed to try get value!");
        return rc;
      }
      if (rc == RC::RECORD_EOF) {
        // 这里表示子查询没有结果
        reg_value.set_null_flag(true);
      }
      if (reg_value.attr_type() != reg_field_meta.type()) {
        // 如果reg_value是NULL的话，就不需要进行类型转换了，直接传入就可以，后续还有处理手段
        if (reg_value.is_null() && reg_field_meta.nullable()) {
          // 什么也不需要做
        }
        // 下面是参数不合理的情况
        else if (!reg_field_meta.nullable() && reg_value.is_null()) {
          LOG_WARN("Field %s is not nullable", reg_field_meta.name());
          return RC::INVALID_ARGUMENT;
        }

        else {
          // 这里就是类型不符合，但是尝试类型转换
          if (INT32_MAX == DataType::type_instance(reg_value.attr_type())->cast_cost(reg_field_meta.type())) {
            LOG_WARN("argument do not match ");
            return RC::INVALID_ARGUMENT;
          } else {
            Value tmp1 = reg_value;
            Value tmp2;
            DataType::type_instance(reg_value.attr_type())->cast_to(tmp1, reg_field_meta.type(), tmp2);
            reg_value = tmp2;
          }
        }
      }
      field_meta.push_back(reg_field_meta);
      value.push_back(reg_value);
    }
    LOG_TRACE("--------cells_size:%d", cells.size());
    rc = table_->make_new_record(static_cast<int>(cells.size()), cells.data(), new_record, field_meta, value);
    if (rc != RC::SUCCESS) {
      LOG_WARN("failed to make new record!");
      return rc;
    }
    Record &old_record = row_tuple->record();
    old_records_.emplace_back(std::move(old_record));
    new_records_.push_back(new_record);
  }
  LOG_TRACE("----------------------RECORD OK");
  child->close();
  // 现在new_records中存放着需要被插入的数据，old_records中存放着需要被删除的数据

  // 先删除再插入
  // 记录的有效性由事务来保证，如果事务不保证删除的有效性，那说明此事务类型不支持并发控制，比如VacuousTrx
  //
  // 下面的两个数组大小都是0，上面的代码段一定是有问题的
  LOG_TRACE("old_record_size:%d", old_records_.size());
  LOG_TRACE("new_record_size:%d", new_records_.size());
  rc = RC::SUCCESS;
  for (Record &record : old_records_) {
    rc = trx_->delete_record(table_, record);
    if (rc != RC::SUCCESS) {
      LOG_WARN("failed to delete record: %s", strrc(rc));
      return rc;
    }
  }
  // 插入
  for (Record &record : new_records_) {
    rc = trx_->insert_record(table_, record);
    if (rc != RC::SUCCESS) {
      LOG_WARN("failed to insert record: %s", strrc(rc));
      return rc;
    }
  }

  // 这里手动释放了value_的资源
  for (unsigned i = 0; i < value_.size(); i++) {
    value_[i].reset();
  }

  return rc;
}

RC UpdatePhysicalOperator::next() { return RC::RECORD_EOF; }

RC UpdatePhysicalOperator::close() { return RC::SUCCESS; }
